package com.ir.stat.spark.service;


import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.ir.stat.spark.utils.JsonBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class StatService {

    private Logger logger = LoggerFactory.getLogger(StatService.class);

    private Queue<StatTask> taskQueue = new LinkedBlockingDeque();

    private ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 200, TimeUnit.MILLISECONDS,
            new LinkedBlockingDeque());

    private Cache<String, String> resultMap = CacheBuilder.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build();

    @PostConstruct
    private void boot() {
        new Thread(() -> {
            while (true) {
                try {
                    StatTask task;
                    while ((task = taskQueue.poll()) != null) {
                        logger.info("executing task {}", task);
                        task.setListener((String resourceId, String result, boolean isError, long elapsed) -> {
                            logger.info("task = {} finished, isError = {}, result = {} ", resourceId, isError, result);
                            Map ret = new HashMap();
                            ret.put("isError", isError);
                            ret.put("result", result);
                            ret.put("elapsed", elapsed);
                            ret.put("resourceId", resourceId);
                            String json = JsonBuilder.getInstance().toJson(ret);
                            resultMap.put(resourceId, json);
                        });
                        executor.execute(task);
                    }
                    Thread.sleep(1);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public boolean offerTask(Map params) {
        StatTask task = new StatTask(params);
        if (resultMap.getIfPresent(task.getResourceId()) != null)
            return false;
        return taskQueue.offer(task);
    }

    public Map getTaskResult(String key) {
        Map ret = new HashMap();
        String json = resultMap.getIfPresent(key);
        if (json != null) {
            Map map = JsonBuilder.getInstance().fromJson(json, HashMap.class);
            ret.putAll(map);
        }
        return ret;
    }
}
